package org.apache.ocean;

import java.io.File;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.lucene.index.CorruptIndexException;
import org.apache.ocean.util.OceanConcurrentHashMap;
import org.apache.ocean.util.WeakRefManager;

/**
 * 
 * @author jason rutherglen
 */
public class SearchIndexShardManager {
	public static Logger log = Logger.getLogger(SearchIndexShardManager.class.getName());
	private ConcurrentHashMap<String, SearchIndexShard> searchIndexShardMap = new OceanConcurrentHashMap<String, SearchIndexShard>();
	private ReentrantReadWriteLock searchIndexShardLock = new ReentrantReadWriteLock();
	private SearchService searchService;
	private SearchIndexShard mainSearchDirectory;
	private SearchIndexShardPool writeableShardPool;
	private WeakRefManager<SearchIndexShard> weakRefManager = new WeakRefManager<SearchIndexShard>();
	private int minWriteableShards = 4;

	public SearchIndexShardManager(SearchService searchService) throws Exception {
		this.searchService = searchService;
		init();
	}
	
	public void createShards(Map<String,ShardInfo> shardInfos) throws Exception {
		//log.info("shardInfos: "+shardInfos);
		for (ShardInfo shardInfo : shardInfos.values()) {
			if (!searchIndexShardMap.containsKey(shardInfo.id)) {
				createSearchIndexShard(shardInfo.id);
			}
		}
	}
	
	public class CreateSearchIndexShardsTask implements Runnable {
		int num;

		public CreateSearchIndexShardsTask(int num) {
			this.num = num;
		}

		public void run() {
			log.info("creating " + num + " shards");
			for (int x = 0; x < num; x++) {
				try {
					SearchIndexShard searchIndexShard = createSearchIndexShard();
					addShardNoLock(searchIndexShard);
					// searchService.sendNameUpdateNoWait(searchIndexShard.getNodeShard());
				} catch (Throwable throwable) {
					log.log(Level.SEVERE, "", throwable);
				}
			}
		}
	}

	public int getMinWriteableShards() {
		return minWriteableShards;
	}

	public void resetSequences() throws Exception {
		List<SearchIndexShard> searchIndexShards = getSearchIndexShards();
		for (SearchIndexShard searchIndexShard : searchIndexShards) {
			searchIndexShard.resetDocumentIDSequence();
			searchIndexShard.resetTransactionIDSequence();
		}
	}

	//public Runnable getShardDeleteTask() {
	//	return new ShardDeleteTask();
	//}

	/**
	 * Delete shards that are no longer referenced.
	 
	public class ShardDeleteTask implements Runnable {
		public void run() {
			try {
				WeakRefManager.Ref<SearchIndexShard> ref = weakRefManager.poll(1000);
				if (ref != null) {
					SearchIndexShard searchIndexShard = ref.get();
					if (searchIndexShard.isClosed()) {
						File directoryFile = searchIndexShard.getDirectoryFile();
						FileUtils.deleteDirectory(directoryFile);
					}
					ref.remove();
				}
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
			}
		}
	}
  */
	public SearchIndexShard getSearchIndexShard(String name) {
		return searchIndexShardMap.get(name);
	}

	public List<SearchIndexShard> getSearchIndexShards() {
		return new ArrayList<SearchIndexShard>(searchIndexShardMap.values());
	}

	public SearchService getSearchService() {
		return searchService;
	}

	public void close() {
		LockUtil.lock(searchIndexShardLock.writeLock());
		try {
			Collection<SearchIndexShard> values = searchIndexShardMap.values();
			for (SearchIndexShard searchIndexShard : values) {
				try {
					searchIndexShard.close();
				} catch (Throwable throwable) {
					log.log(Level.SEVERE, "", throwable);
				}
			}
		} finally {
			searchIndexShardLock.writeLock().unlock();
		}
	}

	public class ShardHandler implements SearchIndexShard.ShardInterface {
		public void newTransactionRecord(TransactionRecord transactionRecord) throws Exception {
			searchService.newTransactionRecord(transactionRecord);
		}
		
		public CurrentIndexSchema getCurrentIndexSchema() throws Exception {
			return searchService.getCurrentIndexSchema();
		}

		public String getCellID() throws Exception {
			return searchService.getCellIDNoWait();
		}

		public void reportCorruptIndex(SearchIndexShard searchIndexShard, CorruptIndexException corruptIndexException) {
			searchService.reportCorruptIndex(searchIndexShard, corruptIndexException);
		}

		public void shardClosed(SearchIndexShard searchIndexShard) {
			try {
				removeShard(searchIndexShard);
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
			}
		}
	}

	void init() throws Exception {
		writeableShardPool = new SearchIndexShardPool(ClusterConstants.READ_WRITE, getByType(ClusterConstants.READ_WRITE), this);
		// load the existing directories
		File shardsDirectory = searchService.searchConfig.indicesDirectory;
		File[] shardDirectories = shardsDirectory.listFiles();
		for (File shardDirectory : shardDirectories) {
			if (shardDirectory.isDirectory()) {
				SearchIndexShard searchDirectory = new SearchIndexShard(shardDirectory, new ShardHandler());
				addShardNoLock(searchDirectory);
			}
		}
		//int toCreate = getMinWriteableShards() - writeableShardPool.size();
		//if (toCreate > 0) {
		//	new CreateSearchIndexShardsTask(toCreate).run();
		//}
		//writeableShardPool.getNext();
	}
  
	public void createMinimumNumberOfShards() {
		int toCreate = getMinWriteableShards() - writeableShardPool.size();
		if (toCreate > 0) {
			new CreateSearchIndexShardsTask(toCreate).run();
		}
	}
	
	// TODO: make sure shard is properly created, have log of creation of shard, if it is never completely created then delete it
	void addShardNoLock(SearchIndexShard searchIndexShard) throws Exception {
		searchIndexShard.setType(ClusterConstants.READ_WRITE);
		log.log(Level.INFO, "addSearchDirectoryNoLock: " + searchIndexShard);
		if (!searchIndexShardMap.contains(searchIndexShard.getID())) {
			searchIndexShardMap.put(searchIndexShard.getID(), searchIndexShard);
		}
		if (searchIndexShard.typeEquals(ClusterConstants.READ_WRITE)) {
			writeableShardPool.add(searchIndexShard);
		}
		weakRefManager.add(searchIndexShard);
	}

	void removeShard(SearchIndexShard searchDirectory) throws Exception {
		log.log(Level.INFO, "removeSearchDirectory: " + searchDirectory);
		Object removedSearchDirectory = searchIndexShardMap.remove(searchDirectory.getID());
		log.log(Level.INFO, "removedSearchDirectory: " + (removedSearchDirectory != null));
	}

	public void changeSearchIndexShardType(SearchIndexShard searchIndexShard, String newType) throws Exception {
		searchIndexShard.setType(newType);
		// take it out of the write queue
		if (newType.equals(ClusterConstants.READ)) {
			log.log(Level.INFO, "writeableDirectoryQueue.remove: " + searchIndexShard);
			writeableShardPool.remove(searchIndexShard.getID());
		}
	}

	private SearchIndexShard createSearchIndexShard(String id) throws Exception {
		File indicesDirectory = searchService.searchConfig.indicesDirectory;
		File shardDirectory = new File(indicesDirectory, id);
		SearchIndexShard searchIndexShard = new SearchIndexShard(shardDirectory, new ShardHandler());
		addShardNoLock(searchIndexShard);
		return searchIndexShard;
	}
	
	SearchIndexShard createSearchIndexShard() throws Exception {
		java.util.UUID uuid = java.util.UUID.randomUUID();
		String id = uuid.toString();
		return createSearchIndexShard(id);
	}

	public SearchIndexShard getNextWriteableIndexShard() throws Exception {
		log.info("writeableShardPool.size: "+writeableShardPool.size());
		return writeableShardPool.getNext();
	}

	public List<SearchIndexShard> getByType(String status) {
		try {
			Collection<SearchIndexShard> values = searchIndexShardMap.values();
			List<SearchIndexShard> list = new ArrayList<SearchIndexShard>(searchIndexShardMap.size());
			for (SearchIndexShard searchDirectory : values) {
				if (searchDirectory.getType().equals(status)) {
					list.add(searchDirectory);
				}
			}
			return list;
		} finally {
		}
	}
}
